Add reliable message queues - #451
Draft
zzstoatzz wants to merge 9 commits into
Draft
Conversation
❌ 1 Tests Failed:
View the top 1 failed test(s) by shortest run time
To view more test analytics, go to the Test Analytics Dashboard |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
this PR adds a small Redis Streams queue primitive for runtimes that need Docket reliability without delegating execution or results to Docket.
This is motivated by PrefectHQ/prefect#21218: Prefect owns task execution, state, and result persistence, but its API replicas need a shared, crash-recoverable delivery layer.
API and semantics
docket.queue(name)creates a keyed queue.put()publishes opaque bytes with queue-wide idempotency and optional topic backpressure.subscribe()provides competing consumers across FIFO topics, with priority among ready messages.acknowledge()durably deletes accepted work.release()atomically moves work to another topic for immediate retry.Delivery is intentionally at least once. The downstream runtime remains responsible for idempotent execution and results.
Implementation
The queue uses Redis Streams consumer groups,
XAUTOCLAIM, and small atomic Lua operations for publish, acknowledge, and release. Redis keys, clients, group recovery, visibility renewal, and deduplication storage remain private Docket details.One subscription uses a constant two background tasks and a single multi-stream reader regardless of topic count. Every claimed message—including Redis-side prefetch—is visibility-renewed, and graceful shutdown releases internal buffers immediately. Acknowledged consumers can advance while the durable Redis acknowledgement completes; if that write fails, the original stream entry remains and is redelivered.
The implementation reuses Docket existing standalone, cluster, Sentinel, ACL, and in-memory connection paths.
Validation
XAUTOCLAIM.uv run prek run --all-filespasses, including Ruff, file-size limits, source/test Pyright, and 100% public type completeness.0.0.0fallback version, allowing immutable HTTPS pins in slim images without Git._lua.py,_redis.py,_concurrency.py, and worker tests; no queue line or branch is missing.Compatibility
Existing task scheduling and result APIs are unchanged. Queue use requires Redis 6.2 or newer; Docket task execution continues to require only Redis Streams support.